package net.bwie.realtime.jtp.ods.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;

public class CdcDataEtlMapFunction extends RichMapFunction< String, String>
{
    @Override
    public String map(String value) throws Exception {
        // 解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        String afterValue = jsonObject.getString("after");
        String beforeValue = jsonObject.getString("before");
        String opValue = jsonObject.getString("op");
        Long tsMs = jsonObject.getLong("ts_ms");
        JSONObject object = jsonObject.getJSONObject("source");
        String dbValue = object.getString("db");
        String tableValue = object.getString("table");
        if (tableValue.equals("user_info")){
            System.out.println(tableValue);
        }
        String operatorValue;
        switch (opValue){
             case "c":
                operatorValue = "insert";
                break;
            case "u":
                operatorValue = "update";
                break;
            case "d":
                operatorValue = "delete";
                break;
            default:
                operatorValue = "unknown";
                break;
        }
         JSONObject newOvject = new JSONObject();
         newOvject.put("operate_type", operatorValue);
         newOvject.put("db_name", dbValue);
         newOvject.put("table_name", tableValue);
         newOvject.put("ts_ms", tsMs);
         if (opValue.equals("d")){
             newOvject.put("operate_date", JSON.parseObject(beforeValue));
         }else {
             newOvject.put("operate_date", JSON.parseObject(afterValue));
         }
        return newOvject.toJSONString();
    }
}
